package com.audaque.springboot.foshanupload.reactor.webfluxdemo.controller;

import lombok.extern.slf4j.Slf4j;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.http.codec.multipart.Part;
import org.springframework.util.MultiValueMap;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.io.IOException;
import java.net.URI;
import java.nio.CharBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @author zgb
 * @desc ...
 * @date 2023-08-02 17:00:15
 */
@Slf4j
@RestController
@RequestMapping("webflux")
public class WebfluxController {


    @PostMapping("getParam")
    public Mono<Object> getParam(ServerWebExchange exchange) throws IOException {
        Map<String, String> stringStringMap = exchange.getRequest().getHeaders().toSingleValueMap();
        log.debug("getHeaders:{}", stringStringMap);
        //打印head
        Mono<MultiValueMap<String, String>> formData = exchange.getFormData();
        //获取form-data中的内容
        formData.subscribe(map -> {
            Map<String, String> formMap = map.toSingleValueMap();
            log.debug("getFormData:{}", formMap);
        });
        //如果是获取post或者get请求之类的body的话就从exchange.getRequest().getBody()中获取
        Map<String, String> queryParams = exchange.getRequest().getQueryParams().toSingleValueMap();
        log.debug("getQueryParams:{}", queryParams);
        Flux<DataBuffer> body = exchange.getRequest().getBody();
        log.debug("getBody:{}", body);
        //消费到线程安全原子类
        AtomicReference<String> bodyRef = new AtomicReference<>();
        body.subscribe(buffer -> {
            CharBuffer charBuffer = StandardCharsets.UTF_8.decode(buffer.asByteBuffer());
            DataBufferUtils.release(buffer);
            bodyRef.set(charBuffer.toString());
        });
        String bodyStr = bodyRef.get();
        log.debug("AtomicReference Body：" + bodyStr);

        URI uri = exchange.getRequest().getURI();
        log.debug("getURI:{}", uri);

        exchange.getMultipartData().subscribe(data -> {
                    Map<String, Part> multipartData = data.toSingleValueMap();
                    log.debug("getMultipartData:{}", multipartData);

                    Part part1 = multipartData.get("username");
                    Flux<DataBuffer> content1 = part1.content();
                    AtomicReference<String> ref1 = new AtomicReference<>();
                    content1.subscribe(buffer -> {
                        CharBuffer charBuffer = StandardCharsets.UTF_8.decode(buffer.asByteBuffer());
                        DataBufferUtils.release(buffer);
                        ref1.set(charBuffer.toString());
                    });
                    String username = ref1.get();
                    log.debug("username：" + username);

                }
        );
        return Mono.empty();

    }

    @PostMapping("stream")
    public void TestTrace() {
        Stream.of("beijing", "tianjin", "shanghai", "wuhan")
                .map(String::length)
                .filter(e -> e > 5)
                .collect(Collectors.toList());
    }


}
